package org.dlx.drpc.core.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import org.dlx.drpc.core.common.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

public class RPCClient implements IRPCClient
{
    private static Logger logger = LoggerFactory.getLogger(RPCClient.class);

    private String ip;
    private int port;
    private EventLoopGroup group;
    private Bootstrap bootStrap;
    private IRPCEncoder rpcEncoder;
    private IRPCDecoder rpcDecoder;

    public RPCClient(String ip, int port, IRPCEncoder en, IRPCDecoder de)
    {
        this.ip = ip;
        this.port = port;
        this.rpcEncoder = en;
        this.rpcDecoder = de;
    }
    @Override
    public RPCClient start() throws Exception
    {
        group = new NioEventLoopGroup();
        bootStrap = new Bootstrap();
        bootStrap.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>()
                {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception
                    {
                        ChannelPipeline pipe = ch.pipeline();
                        //解密添加长度解析
                        pipe.addLast(new LengthFieldBasedFrameDecoder(65536, 0, 4, 0,4));
                        //加密添加一个长度字段
                        pipe.addLast(new LengthFieldPrepender(4, false));
                        pipe.addLast(new MessageDecoder());//将bytebuf编码成message
                        pipe.addLast(new MessageEncoder());//将Message 编码成bytebuf
                        pipe.addLast(new RPCDecoder(rpcDecoder));
                        pipe.addLast(new RPCEncoder(rpcEncoder));
                        pipe.addLast(new MessagePromisor());
                    }
                });
        return this;
    }
    @Override
    public RPCClient close()
    {
        group.shutdownGracefully();
        return this;
    }
    //把链接对象暴露出去方便在外面做连接池,因为RPC通信可能会在返回数据的格式及内容
    //不正确时认为链接不可用需要关闭连接 在协议自定义 无法检测协议内部数据的情况下
    // 内部做连接池可能会在极端情况下出问题
    public RPCConn newConn(){
        return new RPCConn(bootStrap.connect(this.ip, this.port), false);
    }
    //每次都会新建链接并且关闭
    @Override
    public <T> T send(Object payload, Class<T> cla, long ms) throws Exception {
        return send(new RPCConn(bootStrap.connect(this.ip, this.port)), payload, cla, ms);
    }
    //T re = (T) pro.getResult() 类型不对 不会触发异常,如果支持instance of T就好了
    //添加一个cla参数来验证返回类型
    @Override
    public <T> T send(RPCConn conn, Object payload, Class<T> cla, long ms) throws Exception
    {
        try{
            ChannelFuture f = conn.conn.sync();
            Channel  ch= null;
            ch = f.channel();
            Promise pro = new Promise(ch.newPromise());
            ch.attr(AttributeMapKeys.RPC_PROMISE).set(pro);
            ch.writeAndFlush(payload);
            pro.getPromise().get(ms, TimeUnit.MILLISECONDS);
            Object ob = pro.getResult();
            if(!ob.getClass().equals(cla))
                throw new Exception("返回类型出错:" + ob.getClass().toString());
            @SuppressWarnings("unchecked")
            T re = (T)ob;
            return re;
        }finally
        {
            if(conn.owner)
                conn.conn.channel().pipeline().close().sync();
        }
    }

    @Override
    public String getAddress(){
        return ip + ":" + Integer.toString(this.port);
    }
}
